/**
 * AbstractTrafficAggregator.java 03.07.2006
 */
package org.dicr.traffic.aggregator;

import java.util.*;

import org.dicr.traffic.source.*;
import org.slf4j.*;

/**
 * Abstract Traffic Aggregator.
 * 
 * @author <A href='http://dicr.org'>Igor A Tarasov</A>
 * @version 060703
 */
public abstract class AbstractTrafficAggregator extends AbstractTrafficSource implements TrafficAggregator {

	private static final Logger log = LoggerFactory.getLogger(AbstractTrafficAggregator.class);

	private final Collection<TrafficSource> sources = new ArrayList<>();

	private final TrafficListener trafficListener = new AggregatorListener();

	private final List<TrafficElement> buffer = new ArrayList<>();

	private Timer flushTimer = null;

	/** count of elements, received by aggregator from source */
	private long receivedElementsCount = 0;

	/** Buffer size */
	private int bufferSize = 1000;

	/** Buffer flush period in seconds */
	private long flushPeriod = 60;

	/**
	 * Add new source.
	 * 
	 * @param source
	 *            source to use in aggregation
	 * @see org.dicr.traffic.aggregator.TrafficAggregator#addSource(org.dicr.traffic.source.TrafficSource)
	 */
	@Override
	public final void addSource(final TrafficSource source) {
		if (source == null) throw new IllegalArgumentException("null source");
		synchronized (this.sources) {
			if (this.sources.contains(source)) return;
			source.registerListener(this.trafficListener);
			this.sources.add(source);
		}
		AbstractTrafficAggregator.log.debug("added source: " + source);
	}

	/**
	 * Remove source.
	 * 
	 * @param source
	 *            source, to remove from aggregation
	 * @see org.dicr.traffic.aggregator.TrafficAggregator#removeSource(org.dicr.traffic.source.TrafficSource)
	 */
	@Override
	public final void removeSource(final TrafficSource source) {
		if (source == null) throw new IllegalArgumentException("null source");
		synchronized (this.sources) {
			if (!this.sources.contains(source)) return;
			source.unregisterListener(this.trafficListener);
			this.sources.remove(source);
		}
		AbstractTrafficAggregator.log.debug("removed source: " + source);
	}

	/**
	 * Set sources. Previous sources will remove from aggregation.
	 * 
	 * @param aSources
	 *            sources to use in aggregation
	 * @see org.dicr.traffic.aggregator.TrafficAggregator#setSources(java.util.Collection)
	 */
	@Override
	public final void setSources(final Collection<TrafficSource> aSources) {
		if (aSources == null) throw new IllegalArgumentException("null sources");
		synchronized (this.sources) {
			for (final TrafficSource source : this.sources) {
				source.unregisterListener(this.trafficListener);
			}
			this.sources.clear();
			for (final TrafficSource source : aSources) {
				source.registerListener(this.trafficListener);
				this.sources.add(source);
			}
		}
		AbstractTrafficAggregator.log.info("configured " + aSources.size() + " sources");
	}

	/**
	 * Return sources.
	 * 
	 * @return sources, which currently used in aggregation.
	 * @see org.dicr.traffic.aggregator.TrafficAggregator#getSources()
	 */
	@Override
	public final Collection<TrafficSource> getSources() {
		synchronized (this.sources) {
			return new ArrayList<>(this.sources);
		}
	}

	/**
	 * Return buffer size.
	 * 
	 * @return current buffer size
	 * @see org.dicr.traffic.aggregator.TrafficAggregator#getBufferSize()
	 */
	@Override
	public final int getBufferSize() {
		synchronized (this.buffer) {
			return this.bufferSize;
		}
	}

	/**
	 * Set buffer size.
	 * 
	 * @param size
	 *            new buffer size.
	 */
	@Override
	public final void setBufferSize(final int size) {
		if (size < 1) throw new IllegalArgumentException("size: " + size);
		synchronized (this.buffer) {
			this.bufferSize = size;
		}
		AbstractTrafficAggregator.log.debug("configured buffer size: " + size);
	}

	/**
	 * Return flush period.
	 * 
	 * @return period in seconds to flush buffer to listeners
	 */
	public final long getFlushPeriod() {
		return this.flushPeriod;
	}

	/**
	 * Set flush period.
	 * 
	 * @param period
	 *            period in seconds to flush buffer to listeners or <= 0 to
	 *            disable flushing
	 */
	public final void setFlushPeriod(final long period) {
		synchronized (this.buffer) {
			this.flushPeriod = period;
			if (this.flushTimer != null) {
				this.flushTimer.cancel();
				this.flushTimer = null;
			}
			if (period > 0) {
				this.flushTimer = new Timer("Traffic aggregator flush", false);
				this.flushTimer.scheduleAtFixedRate(new FlushTask(), period * 1000, period * 1000);
				AbstractTrafficAggregator.log.info("configured buffer flush period: " + period);
			} else AbstractTrafficAggregator.log.info("buffer flushhing disabled");
		}
	}

	/**
	 * Return count of aggregated elements.
	 * 
	 * @return total count of elements, which aggregated with other elements in
	 *         buffer
	 */
	public final long getReceivedElementsCount() {
		return this.receivedElementsCount;
	}

	/**
	 * Increase aggregated elements count.
	 */
	private final void incReceivedElementsCount(final long count) {
		if (count < 0) throw new IllegalArgumentException("count: " + count);
		synchronized (this.buffer) {
			this.receivedElementsCount += count;
			if (this.receivedElementsCount < 0) this.receivedElementsCount = 0;
		}
	}

	/**
	 * Reset counters.
	 */
	@Override
	public final void resetCounters() {
		synchronized (this.buffer) {
			super.resetCounters();
			this.receivedElementsCount = 0;
		}
	}

	/**
	 * Return aggregation coefficient. Aggregation coefficient calculated as:
	 * 
	 * <PRE>
	 * aggregationCoefficient = producedElementsCount / receivedElementsCount
	 * </PRE>
	 * 
	 * @return coefficient of aggregation (count of produced elements divided by
	 *         count of received elements) or 0 if no
	 *         yet elements received by aggregator.
	 */
	public final float getAggregationCoefficient() {
		synchronized (this.buffer) {
			if (this.receivedElementsCount == 0) return 0;
			return (float) this.getProducedElementsCount() / (float) this.receivedElementsCount;
		}
	}

	/**
	 * Do aggregation of element. Element aggregated or added to buffer. No
	 * buffer size checks need to be performed.
	 * 
	 * @param theBuffer
	 *            buffer to aggregate to
	 * @param element
	 *            element to aggregate or add to buffer.
	 */
	protected abstract void aggregateElement(List<TrafficElement> theBuffer, TrafficElement element);

	/**
	 * Flush buffer to listeners.
	 */
	protected final void flushBuffer() {
		synchronized (this.buffer) {
			this.fireTraffic(new ArrayList<>(this.buffer));
			this.buffer.clear();
			AbstractTrafficAggregator.log.trace("buffer flushed, aggregation coefficient: "
					+ this.getAggregationCoefficient());
		}
	}

	/**
	 * Aggregate traffic element to buffer. If buffer size if greater then
	 * bufferSize, fire TrafficEvent.
	 * 
	 * @param element
	 *            element to aggregate
	 */
	protected void aggregate(final TrafficElement element) {
		if (element == null) throw new IllegalArgumentException("null element");
		synchronized (this.buffer) {
			// do aggregation to buffer
			this.aggregateElement(this.buffer, element);
			// update aggregation time
			if (this.buffer.size() >= this.bufferSize) this.flushBuffer();
		}
	}

	/**
	 * Aggregate traffic to buffer.
	 * 
	 * @param traffic
	 *            traffic to aggregate
	 */
	protected void aggregate(final Collection<TrafficElement> traffic) {
		if (traffic == null) throw new IllegalArgumentException("null traffic");
		for (final TrafficElement el : traffic) {
			this.aggregate(el);
		}
		this.incReceivedElementsCount(traffic.size());
	}

	/**
	 * Listener of sources.
	 */
	protected final class AggregatorListener implements TrafficListener {

		/** Constructor */
		public AggregatorListener() {
			super();
		}

		/**
		 * Process traffic event
		 * 
		 * @param e
		 *            event to process
		 */
		@Override
		public final void processTrafficEvent(final TrafficEvent e) {
			if (e == null) throw new IllegalArgumentException("null event");
			AbstractTrafficAggregator.this.aggregate(e.getTraffic());
		}
	}

	/**
	 * Flush buffer periodic job.
	 */
	protected final class FlushTask extends TimerTask {

		/** Constructor */
		public FlushTask() {
			super();
		}

		/** Flush buffer to listeners */
		@Override
		public final void run() {
			AbstractTrafficAggregator.this.flushBuffer();
		}
	}
}
